package com.test.config;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.test.constant.MqttConstants;
import com.test.listener.RocketmqTestListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

/**
 * @author ssang
 * @date 2022-02-11 15:55
 * 初始化生产者、消费者
 */
@Configuration
public class MqttManager {
    @Autowired
    AliMqttConfig mqttProperties;

    /**
     * 生产者
     * @return
     */
    @Bean("mqttProducer")
    public Producer getProducer(){
        Producer producer;
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.ProducerId, mqttProperties.getProducerId());
        properties.put(PropertyKeyConst.AccessKey, mqttProperties.getAccessKey());
        properties.put(PropertyKeyConst.SecretKey, mqttProperties.getSecretKey());
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, mqttProperties.getSendMsgTimeOutMillis());
        properties.put(PropertyKeyConst.ONSAddr, mqttProperties.getOnsAddr());
        producer = ONSFactory.createProducer(properties);

        // 在发送消息前，必须调用 start 方法来启动 Producer，只需调用一次即可
        if(producer.isClosed()){
            producer.start();
        }

        return producer;
    }

    /**
     * 消费者
     * 需要调用getConsumer方法获取，保证获得的consumer是new出来的实例
     * 调用者获取consumer后自己subscribe和start
     */
//    @Bean("mqttConsumer")
    public Consumer getConsumer(){
        Consumer consumer;
        Properties properties = new Properties();
        properties.setProperty(PropertyKeyConst.ConsumerId, mqttProperties.getConsumerId());
        properties.setProperty(PropertyKeyConst.AccessKey, mqttProperties.getAccessKey());
        properties.setProperty(PropertyKeyConst.SecretKey, mqttProperties.getSecretKey());
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, mqttProperties.getSendMsgTimeOutMillis());
        properties.setProperty(PropertyKeyConst.ONSAddr, mqttProperties.getOnsAddr());
        consumer = ONSFactory.createConsumer(properties);

/*
        //订阅
        consumer.subscribe(MqttConstants.MQTT_QUEUE_TOPIC, MqttConstants.MQTT_QUEUE_TAG, new RocketmqTestListener());//监听第一个topic，new对应的监听器

        // 在发送消息前，必须调用start方法来启动consumer，只需调用一次即可，当项目关闭时，自动shutdown
        if(consumer.isClosed()){
            consumer.start();
        }

 */

        return consumer;
    }
}
